ヒューマンインザループとは？
ヒューマンインザループ（Human-in-the-Loop）とは、AI エージェントのツール実行前に人間が確認できる仕組みです。
例えば、エージェントがファイルへの書き込みや SQL を実行する前に、人間に確認を求めることができます。問題なければ「承認」、内容を変えたければ「修正」、不要であれば「却下」という判断を人間が行います。
本記事では、ヒューマンインザループの概要を説明し、Python のサンプルプログラムを実装します。
LangChain によるヒューマンインザループ
さっそく LangChain でヒューマンインザループ機能を持ったエージェントを構築していきましょう。
OpenAI や Anthropic などのAPIキーをあらかじめご準備ください。
実行には Python 3 の環境が必要です。Google Colab 等でお試しください。Google Colab でコマンドを実行する際は先頭に ! を付けて実行してください。
Python のバージョン
|
1 2 |
$ python --version Python 3.12.12 |
パッケージのインストール
利用したいモデルに応じてパッケージのインストールが必要です。ここでは OpenAI と Anthropic の場合を記載します。
OpenAI のモデルを利用する場合
|
1 |
$ pip install langchain langchain-openai |
Anthropic のモデルを利用する場合
|
1 |
$ pip install langchain langchain-anthropic |
パッケージのバージョンは次の通りです。
|
1 2 3 4 5 6 7 8 9 10 11 12 |
$ pip show langchain langchain-openai langchain-anthropic Name: langchain Version: 1.2.10 ... --- Name: langchain-openai Version: 1.1.10 ... --- Name: langchain-anthropic Version: 1.3.4 ... |
サンプルプログラム
チャットでエージェントを起動する CLI プログラムを実装します。エージェントはユーザー入力に基づいて、ツールの実行を提案します。
ツール実行の提案に対してユーザーが取れる行動は、承認(approve)、修正(edit)、却下(reject)のいずれかです。
-
- 承認(approve) … ツールの実行を承認します。
- 修正(edit) … ツール名や引数を修正して実行します。エージェントが提案したツールと引数が、期待していたものと異なる場合に選択できます。※ サンプルプログラムでは簡略化のため、ツールの引数のみ修正できるようにしています。
- 却下(reject) … ツールの実行を却下します。却下理由をメッセージで伝えることもできます。
ソースコードは次の通りです。10 ～ 14 行目の YOUR_XXXX_API_KEY の箇所を、利用したいモデルの API キーに書き換えてください。
ポイントはソースコード上にコメントしてあります。
LangChain のドキュメントを参考にしております。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 |
import os from langchain.agents import create_agent from langchain.agents.middleware import HumanInTheLoopMiddleware from langgraph.checkpoint.memory import InMemorySaver from langchain.tools import tool from langgraph.types import Command # YOUR_OPENAI_API_KEY の箇所を実際の値に書き換えてください os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY" MODEL = "gpt-4.1" # Claude (Anthropic) を使う場合は下記2行のコメントアウトを外して、値を書き換えてください # os.environ["ANTHROPIC_API_KEY"] = "YOUR_ANTHROPIC_API_KEY" # MODEL = "claude-sonnet-4-6" @tool def write_file(file_path: str, content: str) -> None: """file_path に content を書き込む""" pass @tool def execute_sql(query: str) -> None: """SQL を実行する""" pass @tool def read_data(file_path: str) -> str: """file_path の内容を読み取る""" return "hoge" agent = create_agent( model=MODEL, tools=[write_file, execute_sql, read_data], middleware=[ HumanInTheLoopMiddleware( interrupt_on={ "write_file": True, # すべての決定（approve, edit, reject）を選択可能にします "execute_sql": {"allowed_decisions": ["approve", "reject"]}, # approve と reject のみ選択可能にします（edit は不可） "read_data": False, # 確認を不要にします }, description_prefix="ツールの実行が承認待ちです: ", ), ], # エージェントの状態を保持するチェックポインターを設定します（ここではインメモリに保存しています） # 永続化したい場合は AsyncPostgresSaver 等で DB に保存するとよいでしょう checkpointer=InMemorySaver(), ) config = {"configurable": {"thread_id": "thread_01"}} interrupt = None while True: user_input = input("ユーザー入力（q で終了）: ") input_obj = None # q でプログラムを終了します if user_input == "q": break # エージェントがユーザーの確認待ちで止まっている場合 if interrupt: # y でツール実行を承認します if user_input == "y": input_obj = Command(resume={"decisions": [{"type": "approve"}]}) # e でツール実行を修正します(ツールの引数を修正します) elif user_input == "e": action_request = interrupt[0].value["action_requests"][0] suggested_tool_name = action_request["name"] suggested_tool_args = action_request["args"] new_tool_args = {} for arg_name in suggested_tool_args: new_tool_args[arg_name] = input(f"{arg_name}: ") input_obj = Command(resume={"decisions": [{ "type": "edit", "edited_action": { "name": suggested_tool_name, "args": new_tool_args } }]}) # n でツール実行を却下します elif user_input == "n": input_obj = Command(resume={"decisions": [{"type": "reject"}]}) # メッセージ付きでツール実行を却下します else: input_obj = Command(resume={"decisions": [{ "type": "reject", "message": user_input }]}) interrupt = None else: input_obj = {"messages": [{"role": "user", "content": user_input}]} for mode, chunk in agent.stream( input_obj, config=config, stream_mode=["updates", "messages"] ): # エージェントのイベントごとに処理します # messages: モデルの出力の場合 if mode == "messages": token, metadata = chunk is_model_node = metadata["langgraph_node"] == "model" if is_model_node and token.content: print(token.content, end="") # updates: エージェントの状態更新の場合 elif mode == "updates": if "__interrupt__" in chunk: interrupt = chunk["__interrupt__"] print(f"\n\nInterrupt: {interrupt}") print() |
動作確認してみます。設定した通り、エージェントがツールを実行する前に確認が求められています。これでヒューマンインザループを実現することができました。
おわりに
LangChain のヒューマンインザループ機能を試してみました。AI エージェントを作成したいが、人間の判断を間に挟みたいというケースで手軽に実装できて便利です。
今回のサンプルプログラムでは、CLI からのチャット起点でエージェントを起動する方式でした。より実践的にする場合は、メールやファイル監視などのイベントをトリガーにエージェントを起動し、 Web UI から確認するといった構成も考えられます。これから AI エージェントを開発される方の参考になれば幸いです。
執筆者プロフィール
- tdi デジタルイノベーション技術部
- AWS上で使えるツールの開発やフロントエンドの調査、生成系AIの調査等を行っています。最近は AWS の CDK やサーバーレスに興味があります。
